package com.sunzm.flink.datastream.scala.operator

import org.apache.commons.lang3.StringUtils
import org.apache.flink.streaming.api.functions.co.{CoFlatMapFunction, CoMapFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}


object ScalaConnectDemo {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //socketTextStream这种数据源比较特殊，只有一个并行度
    val dataStream1: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)
    val dataStream2: DataStream[String] = env.socketTextStream("82.156.210.70", 8888)

    val connectedStreams: ConnectedStreams[String, String] = dataStream1.connect(dataStream2)

    val mapedDS: DataStream[String] = connectedStreams.map(new CoMapFunction[String, String, String] {
      override def map1(value: String): String = {
        // 处理流1的数据
        logger.info(s"map1： ${value}")
        value
      }

      override def map2(value: String): String = {
        //处理流2的数据
        // 处理流2的数据
        logger.info(s"map2： ${value}")
        value
      }
    })

    mapedDS.print("Map DataStream")

    val flatMapedDS: DataStream[String] = connectedStreams.flatMap(new CoFlatMapFunction[String, String, String] {
      override def flatMap1(value: String, out: Collector[String]): Unit = {
        if (StringUtils.isNotBlank(value)) {
          logger.info(s"flatMap1 : ${value}")
          //数据不为空才发出，map无法实现这个功能
          out.collect(value)
        } else {
          logger.info("flatMap1 收到空数据")
        }
      }

      override def flatMap2(value: String, out: Collector[String]): Unit = {
        if (StringUtils.isNotBlank(value)) {
          logger.info(s"flatMap2 : ${value}")
          //数据不为空才发出，map无法实现这个功能
          out.collect(value)
        } else {
          logger.info("flatMap2 收到空数据")
        }
      }
    })

    flatMapedDS.print("FlatMap DataStream")

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
